package com.cennavi.opers;

import java.io.IOException;
import java.io.PrintWriter;
import java.net.URLEncoder;
import java.text.SimpleDateFormat;
import java.util.Date;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

import com.cennavi.dbs.ZooKeeperConfigs;

public class PbfExtract extends HttpServlet {

	private static final long serialVersionUID = -1909422936250456023L;


	public PbfExtract() {
		super();
	}

	/**
	 * Destruction of the servlet. <br>
	 */
	public void destroy() {
		super.destroy(); // Just puts "destroy" string in log
		// Put your code here
	}

	/**
	 * The doGet method of the servlet. <br>
	 * 
	 * This method is called when a form has its tag value method equals to get.
	 * 
	 * @param request
	 *            the request send by the client to the server
	 * @param response
	 *            the response send by the server to the client
	 * @throws ServletException
	 *             if an error occurred
	 * @throws IOException
	 *             if an error occurred
	 */
	public void doGet(HttpServletRequest request, HttpServletResponse response)
			throws ServletException, IOException {

		doPost(request, response);
	}

	/**
	 * The doPost method of the servlet. <br>
	 * 
	 * This method is called when a form has its tag value method equals to
	 * post.
	 * 
	 * @param request
	 *            the request send by the client to the server
	 * @param response
	 *            the response send by the server to the client
	 * @throws ServletException
	 *             if an error occurred
	 * @throws IOException
	 *             if an error occurred
	 */
	public void doPost(HttpServletRequest request, HttpServletResponse response)
			throws ServletException, IOException {

		response.setHeader("Access-Control-Allow-Origin", "*");
		response.setHeader("Access-Control-Allow-Methods",
				"POST, GET, OPTIONS, DELETE,PUT");

		PrintWriter out = response.getWriter();

		try {

			String jsonStr = request
					.getParameter("request");
			
			jsonStr = new String(jsonStr.getBytes("ISO-8859-1"),"UTF-8");
			
			JSONObject json = JSONObject.fromObject(jsonStr);

			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
			String taskId = "task_" + sdf.format(new Date());

			String zooms = json.getString("zooms");
			
			StringBuilder sb2 = new StringBuilder();
			
			String task_name= json.getString("task_name").replaceAll(" ", "");
			
			if (task_name.length()==0) task_name ="blank";
			
			
			String[] splitZooms = zooms.split(",");
			
			int start = Integer.parseInt(splitZooms[0]);
			
			int end = Integer.parseInt(splitZooms[1]);
			
			sb2.append(start);
			
			for(int i=start+1;i<=end;i++){
				sb2.append(",");
				sb2.append(String.valueOf(i));
			}

			String sourceName = json.getString("source_name");

			String path = json.getString("path");

			String version = json.getString("version");

			String areaIds = "no";

			if (json.containsKey("area_ids")
					&& json.getJSONArray("area_ids").size() > 0) {
				areaIds = "";
				
				
				JSONArray areas = json.getJSONArray("area_ids");
				
				if (areas.size()>1){
				
					for (int i = 0; i < areas.size() - 1; i++) {
						areaIds += areas.getString(i) + ",";
					}

					areaIds += areas.getString(areas.size()-1);
				
				}else{
					areaIds += areas.getString(areas.size()-1);
				}
				
				
			}

			String conditions = "no";

			if (json.containsKey("conditions") && !"".equals(json.getString("conditions")))
				conditions = json.getString("conditions");

			String zooUrls = ZooKeeperConfigs.props.getProperty("zoo_urls");

			String names = json.getString("names");
			
			StringBuilder sb = null;
			
			
			if (sourceName.split(",").length >0){
				
			}else {
				if ((request.getParameterMap().containsKey("WKT2")) || json.containsKey("WKT1") || json.containsKey("relation")){
					
				}else{
					if (names.split(",").length>1){
						sb = new StringBuilder(
								"spark-submit --master yarn --num-executors "
										+ 50
										+ " --executor-memory 8500m --class com.cennavi.engine.define_pbf --name lunch --jars /var/lib/hadoop-hdfs/lilei/tools.jar,/var/lib/hadoop-hdfs/lilei/spark-csv_2.11-1.3.2.jar,/var/lib/hadoop-hdfs/lilei/commons-csv-1.1.jar /var/lib/hadoop-hdfs/lilei/engine-a.jar ");

						sb.append(sb2.toString() + " ");
						sb.append(names + " ");
						sb.append(sourceName + " ");
						sb.append(path + " ");
						sb.append(areaIds + " ");
						sb.append(taskId + " ");
						sb.append(version + " ");
						sb.append(zooUrls + " ");
						sb.append(task_name + " ");
						sb.append(conditions);
						
						System.out.println(sb.toString());
					}else{
					
					sb = new StringBuilder(
							"spark-submit --master yarn --num-executors "
									+ 50
									+ " --executor-memory 8500m --class com.cennavi.engine.csv2pbf --name lunch --jars /var/lib/hadoop-hdfs/lilei/tools.jar,/var/lib/hadoop-hdfs/lilei/spark-csv_2.11-1.3.2.jar,/var/lib/hadoop-hdfs/lilei/commons-csv-1.1.jar /var/lib/hadoop-hdfs/lilei/engine-a.jar ");

					sb.append(sourceName + " ");
					sb.append(path + " ");
					sb.append(zooUrls + " ");
					sb.append(sb2.toString() + " ");
					
					sb.append(areaIds + " ");

					sb.append(taskId + " ");

					sb.append(version + " ");
					sb.append(task_name + " ");
					sb.append(conditions);
					
					}
				}
			}
			
			
			
			
			

			final String cmd = sb.toString();

			scala.io.Source.fromURL("http://10.20.20.11:8080/maven-web/opers?param="+URLEncoder.encode(cmd,"UTF-8"),"UTF-8");

			out.println("{\"error_code\":0,\"error_msg\":\"\",\"app_id\":\""
					+ taskId + "\"}");

		} catch (Exception e) {
			
			e.printStackTrace();
			
			out.println("{\"error_code\":-1,\"error_msg\":\"" + e.getMessage()
					+ "\"}");
		} finally {
			out.flush();

			out.close();
		}

	}

	/**
	 * Initialization of the servlet. <br>
	 * 
	 * @throws ServletException
	 *             if an error occurs
	 */
	public void init() throws ServletException {
		// Put your code here
	}

}
